package com.hmdp.service.impl;

import cn.hutool.core.bean.BeanUtil;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.hmdp.dto.Result;
import com.hmdp.entity.SeckillVoucher;
import com.hmdp.entity.VoucherOrder;
import com.hmdp.mapper.VoucherOrderMapper;
import com.hmdp.service.ISeckillVoucherService;
import com.hmdp.service.IVoucherOrderService;
import com.hmdp.utils.RedisIDWorker;
import com.hmdp.utils.UserHolder;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RedissonClient;
import org.springframework.aop.framework.AopContext;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * <p>
 * 服务实现类
 * </p>
 *
 * @author 虎哥
 * @since 2021-12-22
 */
@Slf4j
@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {

    @Resource
    private ISeckillVoucherService seckillVoucherService;

    @Resource
    private RedisIDWorker redisIDWorker;

    @Resource
    private RedissonClient redissonClient;

    @Resource
    private StringRedisTemplate stringRedisTemplate;

    // 秒杀优惠券脚本
    private static final DefaultRedisScript<Long> SECKILL_SCRIPT;
    static {
        SECKILL_SCRIPT = new DefaultRedisScript<>();
        SECKILL_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));
        SECKILL_SCRIPT.setResultType(Long.class);
    }

    //当前类的代理对象
    private IVoucherOrderService proxy;

    // 线程池
    private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();

    // 任务
    class VoucherOrderHandler implements Runnable {
        @Override
        public void run() {
            String queueName = "stream.orders";  // 队列名称
            while(true) {
                try {
                    // 1.从消息队列中取出消息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS stream.orders >
                    // 组名为g1,消费者命名为c1
                    // 每次读取一条消息，阻塞2s
                    // 队列名称stream.orders，读取方式：下一条未被消费的消息
                    List<MapRecord<String, Object, Object>> msgs = stringRedisTemplate.opsForStream().read(
                            Consumer.from("g1", "c1"),
                            StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                            StreamOffset.create(queueName, ReadOffset.lastConsumed())
                    );

                    if(msgs == null || msgs.isEmpty()) continue;  // 没有消息，继续循环读取

                    // 2.处理消息，获取消息中的订单数据
                    MapRecord<String, Object, Object> record = msgs.get(0);  // 每次只读一条消息，所以get(0)即可
                    Map<Object, Object> value = record.getValue(); // 就是传递消息时的键值对
                    // lua脚本中订单id的键设置为id，就是方便这里直接复制，因为VoucherOrder中订单id的属性名就叫id
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);

                    // 3.处理这条秒杀订单
                    proxy.createVoucherOrder(voucherOrder);

                    // 4.手动ack， sack stream.orders g1 id，id指的是读取到的这条消息的id
                    stringRedisTemplate.opsForStream().acknowledge(queueName, "g1", record.getId());
                } catch (Exception e) {
                    log.error("处理秒杀订单异常", e);
                    handlePendingList();
                }
            }
        }
    }


    private void handlePendingList() {
        String queueName = "stream.orders";  // 队列名称
        while(true) {
            try {
                // 1.从消息队列中取出消息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS stream.orders 0
                // 组名为g1,消费者命名为c1
                // 每次读取一条消息，不阻塞
                // 队列名称stream.orders，读取方式：pending-list中的第一条消息
                List<MapRecord<String, Object, Object>> msgs = stringRedisTemplate.opsForStream().read(
                        Consumer.from("g1", "c1"),
                        StreamReadOptions.empty().count(1),
                        StreamOffset.create(queueName, ReadOffset.from("0"))
                );

                if(msgs == null || msgs.isEmpty()) break;  // 没有异常消息，不用处理了退出循环

                // 2.处理消息，获取消息中的订单数据
                MapRecord<String, Object, Object> record = msgs.get(0);  // 每次只读一条消息，所以get(0)即可
                Map<Object, Object> value = record.getValue(); // 就是传递消息时的键值对
                // lua脚本中订单id的键设置为id，就是方便这里直接复制，因为VoucherOrder中订单id的属性名就叫id
                VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);

                // 3.处理这条秒杀订单
                proxy.createVoucherOrder(voucherOrder);

                // 4.手动ack sack stream.orders g1 id，id指的是读取到的这条消息的id
                stringRedisTemplate.opsForStream().acknowledge(queueName, "g1", record.getId());
            } catch (Exception e) {
                log.error("处理秒杀订单异常", e);
                try {
                    Thread.sleep(20);   // 避免频繁的处理异常信息
                } catch (InterruptedException ex) {
                    throw new RuntimeException(ex);
                }
            }
        }
    }

    // 类初始化完成后就开始提交任务
    @PostConstruct
    public void init() {
        SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());
    }


    @Transactional
    public void createVoucherOrder(VoucherOrder voucherOrder) {
        // 保存订单到数据库
        save(voucherOrder);
        // 库存减1
        seckillVoucherService.lambdaUpdate()
                .setSql("stock = stock - 1")
                .eq(SeckillVoucher::getVoucherId, voucherOrder.getVoucherId())
                .update();
    }


    /**
     * 抢购优惠券，消息队列版本
     *
     * @param voucherId 优惠券ID
     * @return 订单id
     */
    @Override
    public Result seckillVoucher(Long voucherId) {
        // 用户id
        Long userId = UserHolder.getUser().getId();
        // 订单id还是由全局id生成器生成
        Long orderId = redisIDWorker.nextId("order");
        // 执行秒杀优惠券脚本，判断库存和一人一单资格校验，并发送消息到消息队列
        Long result = stringRedisTemplate.execute(
                SECKILL_SCRIPT,
                Collections.emptyList(),
                voucherId.toString(),
                userId.toString(),
                orderId.toString()
        );
        // result只有0、1、2三种情况
        int r = result.intValue();
        // 库存或一人一单资格校验不通过
        if(r != 0) {
            return Result.fail(r == 1 ? "库存不足" : "系统繁忙，请稍后再试(不允许重复下单)");
        }
        // 主线程获取当前类的代理对象
        proxy = (IVoucherOrderService) AopContext.currentProxy();
        // 返回订单id
        return Result.ok(orderId);
    }




    /**
     * 抢购优惠券,阻塞队列版本
     *
     * @param voucherId 优惠券ID
     * @return 订单id
     */
    /*@Override
    public Result seckillVoucher(Long voucherId) {
        // 用户id
        Long userId = UserHolder.getUser().getId();
        // 执行秒杀优惠券脚本，判断库存和一人一单资格校验
        Long result = stringRedisTemplate.execute(
                SECKILL_SCRIPT,
                Collections.emptyList(),
                voucherId.toString(),
                userId.toString()
        );
        // result只有0、1、2三种情况
        int r = result.intValue();
        // 库存或一人一单资格校验不通过
        if(r != 0) {
            return Result.fail(r == 1 ? "库存不足" : "系统繁忙，请稍后再试(不允许重复下单)");
        }

        // 订单id还是由全局id生成器生成
        Long orderId = redisIDWorker.nextId("order");

        // 将优惠券id、用户id和订单id传给阻塞队列
        VoucherOrder voucherOrder = new VoucherOrder();
        voucherOrder.setId(orderId);
        voucherOrder.setUserId(userId);
        voucherOrder.setVoucherId(voucherId);
        proxy = (IVoucherOrderService) AopContext.currentProxy();  // 当前代理对象
        orderTasks.add(voucherOrder);

        // 返回订单id
        return Result.ok(orderId);
    }*/


    /**
     * 抢购优惠券
     *
     * @param voucherId 优惠券ID
     * @return 订单id
     */
    /*@Override
    public Result seckillVoucher(Long voucherId) {
        // 1.查询优惠券信息
        SeckillVoucher seckillVoucher = seckillVoucherService.getById(voucherId);
        // 2.判断秒杀是否开始
        if (seckillVoucher.getBeginTime().isAfter(LocalDateTime.now())) {
            return Result.fail("秒杀尚未开始");
        }
        // 3.判断秒杀是否结束
        if (seckillVoucher.getEndTime().isBefore(LocalDateTime.now())) {
            return Result.fail("秒杀已经结束");
        }
        // 4.判断库存是否充足
        if (seckillVoucher.getStock() < 1) {
            return Result.fail("库存不足");
        }


        // 5.一人一单
        // 获取分布式锁
        Long userId = UserHolder.getUser().getId();
        // SimpleRedisLock simpleRedisLock = new SimpleRedisLock(stringRedisTemplate, "order:" + userId);   // 锁名称：order:userId，锁的任然是用户
        RLock lock = redissonClient.getLock("lock:order:" + userId);
        boolean isLock = lock.tryLock();
        if(!isLock) {
            return Result.fail("系统繁忙，请稍后重试(请勿重复下单)");
        }
        try {
            IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
            return proxy.createVoucherOrder(voucherId);
        } finally {
            // 记得释放锁
            lock.unlock();
        }
    }*/



    /*@Transactional
    public Result createVoucherOrder(Long voucherId) {
        // 5.一人一单
        // 5.1查询同一用户同一优惠券的订单数量
        Long userId = UserHolder.getUser().getId();
        Long count = lambdaQuery()
                .eq(VoucherOrder::getUserId, userId)
                .eq(VoucherOrder::getVoucherId, voucherId)
                .count();
        // 5.2同一用户已经抢购过该优惠券，不能再抢购了
        if(count > 0) {
            return Result.fail("您已经抢购过该优惠券了");
        }

        // 6.扣减库存
        boolean isSuccess = seckillVoucherService.lambdaUpdate()
                .setSql("stock = stock - 1")
                .eq(SeckillVoucher::getVoucherId, voucherId)
                .gt(SeckillVoucher::getStock, 0)   // 乐观锁
                .update();
        if (!isSuccess) {
            return Result.fail("系统错误，请重新退出重试");
        }
        // 7.创建订单
        VoucherOrder voucherOrder = new VoucherOrder();
        // 7.1设置id
        Long orderId = redisIDWorker.nextId("order");
        voucherOrder.setId(orderId);
        // 7.2设置下单用户id
        voucherOrder.setUserId(userId);
        // 7.3设置优惠券id
        voucherOrder.setVoucherId(voucherId);
        // 7.4插入优惠券订单表
        save(voucherOrder);
        // 8.返回订单id
        return Result.ok(orderId);
    }*/
}
